import asyncio
import math
import random


class AsyncProducerConsumer:
    def __init__(self, queue_size=10):

        # 创建一个长度为10异步队列
        self.queue = asyncio.Queue(queue_size)
        self.producer = []
        self.consumer = []

    # 生产数据
    async def produce(self, name):
        # 模拟生产者生产数据
        await asyncio.sleep(1)
        for i in range(5):

            time = random.random() * 1
            await asyncio.sleep(time)

            item = {
                "no": f"A0001{i}",
                "name": f"item-{i}",
                "age": math.floor(random.random() * 100),
            }

            await self.queue.put(item)

        print(f"生产者 {name} 生产了: {item} (队列大小: {self.queue.qsize()})")

    # 消费数据
    async def consume_data(self, name):
        """消费者"""
        while True:
            try:
                item = await asyncio.wait_for(self.queue.get(), timeout=2.0)

                time = random.random() * 1
                await asyncio.sleep(time)

                print(f"消费者 {name} 消费了: {item} (队列大小: {self.queue.qsize()})")
                self.queue.task_done()

            except asyncio.TimeoutError:
                print(f"消费者 {name} 超时，退出")
                break

    async def main(self):
        # 创建生产者任务
        for i in range(10):
            producer_task_list = asyncio.create_task(self.produce(f"Producer-{i+1}"))
            self.producer.append(producer_task_list)

        # 创建消费者任务
        for i in range(3):
            consumer_task_list = asyncio.create_task(
                self.consume_data(f"Consumer-{i+1}")
            )
            self.consumer.append(consumer_task_list)

        # 等待所有生产者完成
        await asyncio.gather(*self.producer)

        # 等待队列清空
        await self.queue.join()

        # 取消所有消费者任务
        for consumer in self.consumer:
            consumer.cancel()

        # 等待所有消费者完成
        await asyncio.gather(*self.consumer, return_exceptions=True)


# if __name__ == "__main__":
#     async_producer_consumer = AsyncProducerConsumer()
#     asyncio.run(async_producer_consumer.main())


q = asyncio.Queue()


async def producer():
    for i in range(5):
        await q.put(i)

    queue_len = len(q._queue)
    print("生产了:", queue_len, "个产品")


async def consumer():
    while True:
        item = await q.get()

        if item is None:
            break
        print("消费了:", item)
        q.task_done()


async def main():
    prod_task = asyncio.create_task(producer())
    cons_task = asyncio.create_task(consumer())

    await prod_task
    await q.join()
    await q.put(None)  # 发送终止信号给消费者
    await cons_task


if __name__ == "__main__":
    asyncio.run(main())
